package com.atguigu.chapter08;

import com.atguigu.bean.AdsClickLog;
import com.atguigu.util.AnqclnUtil;
import org.apache.flink.api.common.state.*;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * Author: Pepsi
 * Date: 2023/8/14
 * Desc:
 */
public class Flink06_Project_High_Ads {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 1000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);

        SingleOutputStreamOperator<String> main = env
                .readTextFile("input/AdClickLog.csv")
                .map(line -> {
                    String[] data = line.split(",");
                    return new AdsClickLog(
                            Long.valueOf(data[0]),
                            Long.valueOf(data[1]),
                            data[2],
                            data[3],
                            // valueOf和parseLong的区别是返回值类型，前面是包装类，后面是long，建议分开用，减少自动装箱拆箱的性能浪费
                            Long.parseLong(data[4]) * 1000
                    );
                })
                // 因为要统计每个用户对每个广告的点击量，所以keyBy的时候就用userId和AdsId
                .keyBy(log -> log.getUserId() + "_" + log.getAdsId())
                // 分完组后，对数据进行处理
                .process(new KeyedProcessFunction<String, AdsClickLog, String>() {

                    private ValueState<String> yesterdayState;
                    private ValueState<Boolean> isAddedBlackListState;
                    private ReducingState<Long> clickCountState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        // 存放点击量的状态，使用reducing，键控状态，之和键有关，自动聚合，存Long型
                        clickCountState = getRuntimeContext().getReducingState(new ReducingStateDescriptor<Long>("clickCountState", Long::sum, Long.class));
                        // 黑名单的状态，使用普通的键控状态，存放单个值就行，类型Boolean，是否在黑名单中
                        isAddedBlackListState = getRuntimeContext().getState(new ValueStateDescriptor<Boolean>("isAddedBlackListState", Boolean.class));
                        // 存放日期的状态，为了判断是否跨天，如果跨天的话就要清空黑名单和点击量
                        yesterdayState = getRuntimeContext().getState(new ValueStateDescriptor<String>("yesterdayState", String.class));
                    }

                    @Override
                    public void processElement(AdsClickLog value,
                                               Context ctx,
                                               Collector<String> out) throws Exception {
                        // 将元素的时间戳转成年月日的字符串
                        String today = AnqclnUtil.toDate(value.getTimestamp());
                        // 每来一个数据判断一次，如果这个数据的时间戳和状态里面的时间戳一致，说明没有跨天，就不用清空状态，如果不一样说明跨天了，就清空状态
                        if (!today.equals(yesterdayState.value())) {
                            // 先更新状态里面的数据
                            yesterdayState.update(today);
                            // 清空状态
                            clickCountState.clear();
                            isAddedBlackListState.clear();
                        }
                        // 判断这个数据是否被加入黑名单，如果被加入黑名单就不计算他的点击量了，没有加入的话就计算
                        if (isAddedBlackListState.value()==null) {
                            clickCountState.add(1L);
                        }
                        // 定义打印的提示信息
                        String msg = "用户：" + value.getUserId() + " 对广告：" + value.getAdsId() + " 的点击是：" + clickCountState.get();
                        // 如果点击状态里面的点击量大于等于100，加入黑名单
                        if (clickCountState.get() >= 100) {
                            // 判断是不是第一次加入黑名单，如果是第一次就加入黑名单，把数据输出到测输出流中
                            if (isAddedBlackListState.value() == null) {
                                // 加入到侧输出流中，   注意要加{}
                                ctx.output(new OutputTag<String>("blackList"){}, msg + " 超过阈值100，加入黑名单");
                                // 更新黑名单状态
                                isAddedBlackListState.update(true);
                            }
                        } else {
                            // 不大于100直接将提示信息放到流中
                            out.collect(msg);
                        }
                    }
                });

        // 输出正常流中的数据
        main.print("正常");
        // 获得侧输出流中的数据-黑名单中的数据
        main.getSideOutput(new OutputTag<String>("blackList"){}).print("黑名单");


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
